package async

import (
	"context"
	"github.com/ecodeclub/ekit/syncx/atomicx"
	"jk-time/webook/internal/domain"
	"jk-time/webook/internal/repository"
	"jk-time/webook/internal/service/sms"
	"jk-time/webook/pkg/logger"
	"sync/atomic"
	"time"
)

type AsyncSerivce struct {
	svc  sms.Service
	repo repository.RetryRepository
	l    logger.Logger
	//连续超时的个数
	cnt int32

	//阈值
	//连续超过超时这个数字，切换
	threshold int32

	async *atomicx.Value[bool]
}

func NewAsyncSerivce(svc sms.Service, repo repository.RetryRepository, l logger.Logger, threshold int32) *AsyncSerivce {
	res := &AsyncSerivce{svc: svc, repo: repo, l: l, cnt: 0,
		threshold: threshold, async: atomicx.NewValueOf[bool](false)}
	return res
}
func (s *AsyncSerivce) StartAsyncCycle() {
	for {
		s.StartAsync()
	}
}

// StartAsync 异步发送消息
// 这里我们没有设计退出机制，是因为没啥必要
// 因为程序停止的时候，它自然就停止了
// 原理：这是最简单的抢占式调度
func (a *AsyncSerivce) StartAsync() {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	// 抢占一个异步发送的消息，确保在非常多个实例
	// 比如 k8s 部署了三个 pod，一个请求，只有一个实例能拿到
	retry, err := a.repo.PreemptWaitingSMS(ctx)
	cancel()
	switch err {
	case nil: // 执行发送
		// 这个也可以做成配置的
		ctx, cancel = context.WithTimeout(context.Background(), time.Second)
		err = a.svc.Send(ctx, retry.Biz, retry.Args, retry.Numbers...)
		cancel()
		if err != nil {
			a.l.Error("重试异步发送失败,", logger.Int("当前次数", retry.RetryCnt),
				logger.Int("最大次数", retry.RetryCnt), logger.Error(err))
			if retry.RetryCnt < retry.RetryMax-1 {
				return
			}
		}

		res := err == nil
		ctx, cancel = context.WithTimeout(context.Background(), time.Second)
		err = a.repo.ReportScheduleResult(ctx, retry.Id, res)
		cancel()

		if err != nil {
			a.l.Error("执行异步发送短信成功，但是标记数据库失败",
				logger.Error(err),
				logger.Bool("res", res),
				logger.Int64("id", retry.Id))
		}
	case repository.ErrWaitingSMSNotFound:
		// 抢占失败
		time.Sleep(time.Second * 5)
	default:
		// 正常来说应该是数据库那边出了问题，
		// 但是为了尽量运行，还是要继续的
		// 你可以稍微睡眠，也可以不睡眠
		// 睡眠的话可以帮你规避掉短时间的网络抖动问题
		a.l.Error("抢占异步发送短信任务失败",
			logger.Error(err))
		time.Sleep(time.Second * 10)
	}
}
func (a *AsyncSerivce) Send(ctx context.Context, biz string, args []string, numbers ...string) error {
	a.needAsync()
	//time.Sleep(time.Second * 7)
	if a.async.Load() {
		// 需要异步发送，直接转储到数据库
		err := a.repo.Create(ctx, domain.Retry{
			Biz:      biz,
			Args:     args,
			Numbers:  numbers,
			RetryCnt: 0,
			RetryMax: 3,
		})
		return err
	}
	err := a.svc.Send(ctx, biz, args, numbers...)
	switch err {
	case context.DeadlineExceeded:
		atomic.AddInt32(&a.cnt, 1)
	case nil:
		atomic.StoreInt32(&a.cnt, 0)
	default:
	}
	return err
}
func (a *AsyncSerivce) needAsync() {
	// 这边就是你要设计的，各种判定要不要触发异步的方案
	// 1. 基于响应时间的，平均响应时间
	// 1.1 使用绝对阈值，比如说直接发送的时候，（连续一段时间，或者连续N个请求）响应时间超过了 500ms，然后后续请求转异步
	// 1.2 变化趋势，比如说当前一秒钟内的所有请求的响应时间比上一秒钟增长了 X%，就转异步
	// 2. 基于错误率：一段时间内，收到 err 的请求比率大于 X%，转异步

	// 什么时候退出异步
	// 1. 进入异步 N 分钟后
	// 2. 保留 1% 的流量（或者更少），继续同步发送，判定响应时间/错误率

	if atomic.LoadInt32(&a.cnt) < a.threshold {
		return
	}
	time.AfterFunc(time.Second*5, func() {
		a.async.Store(false)
		atomic.StoreInt32(&a.cnt, 0)
	})
	a.async.Store(true)
}
